Summary of Changes

Hello @arunpandianp, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request focuses on optimizing the performance of Dataflow streaming jobs by significantly reducing garbage collection overhead. It achieves this by introducing a robust strategy for reusing ByteStringOutputStream buffers across multiple operations within the WindmillSink and WindmillStateUtil components. This change is crucial for improving the efficiency and stability of high-throughput streaming applications where frequent buffer allocations can lead to performance bottlenecks.

Highlights

• ByteStringOutputStream Reuse: Implemented reuse of ByteStringOutputStream instances in WindmillSink and WindmillStateUtil to reduce object allocation and garbage collection overhead, particularly in performance-critical encoding paths.
• Thread-Local Buffer Management: WindmillStateUtil now utilizes a ThreadLocal with SoftReference to manage ByteStringOutputStream instances, enabling efficient per-thread buffer reuse for encoding state keys.
• Refactored Metadata Encoding: The encodeMetadata method in WindmillSink has been refactored to accept and utilize a reusable ByteStringOutputStream, ensuring the buffer is reset after use via toByteStringAndReset().
• Removed Guava Preconditions: The dependency on Guava's Preconditions.checkState was removed from WindmillSink.
• Potential Logical Error in Stream Check: A logical inversion was introduced when replacing checkState(stream.size() == 0, ...) with if (stream.size() == 0) { throw new IllegalStateException(...) } in WindmillSink.WindmillStreamWriter.encode. The original code asserted that the stream must be empty before encoding, throwing an error if it was not. The new code, however, throws an exception if the stream is empty, which is the expected and desired state. This appears to be a bug that could prevent correct operation.

Code Review

This pull request introduces a good optimization by reusing ByteStringOutputStream buffers to reduce GC overhead in WindmillSink and WindmillStateUtil. The removal of the unconditional stream.toByteString() in the checkState is also a nice performance improvement.

My review includes a few suggestions to further improve the implementation:

• In WindmillStateUtil, the new thread-local caching logic can be made more efficient and robust by improving the reset logic and exception handling to prevent cache corruption.
• In WindmillSink, I've suggested a minor change to make exception handling more consistent and align with best practices.

Overall, these are valuable performance enhancements.

Code Review

This pull request introduces performance optimizations by reusing ByteStringOutputStream buffers to reduce garbage collection overhead. This is achieved by caching the stream in a ThreadLocal in WindmillStateUtil and by passing a reusable stream to encodeMetadata in WindmillSink. Additionally, an unnecessary call to stream.toByteString() is removed from a precondition check. The changes are well-implemented and should improve performance. I have a couple of suggestions regarding exception handling to improve code clarity and robustness.

Code Review

This pull request introduces a performance optimization by reusing ByteStringOutputStream buffers to reduce garbage collection overhead. The changes are implemented in WindmillSink and WindmillStateUtil. In WindmillSink, a stream is now held as a field and reused for encoding metadata and values. In WindmillStateUtil, a ThreadLocal<SoftReference<...>> is used to cache and reuse a stream for encoding state keys. Additionally, an unconditional call to stream.toByteString() within a checkState has been removed to avoid unnecessary object creation. The changes are well-implemented and robust, ensuring that the reusable streams are reset correctly even in case of exceptions. I have one suggestion to simplify exception handling logic in WindmillStateUtil to reduce code duplication.